package com.example.demo_debezium_pgsql.listener;

import com.example.demo_debezium_pgsql.util.PgProperties;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.debezium.config.Configuration;
import io.debezium.connector.postgresql.PostgresConnectorConfig;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import io.debezium.jdbc.JdbcConfiguration;
import io.debezium.relational.RelationalDatabaseConnectorConfig;
import io.debezium.util.LoggingContext;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Map;
import java.util.Properties;

@Component
public class PGLog implements CommandLineRunner {

    @Autowired
    private PgProperties pgProperties;
    private ObjectMapper objectMapper = new ObjectMapper();

    private static final String DATABASE_CONFIG_PREFIX = "database.";
    private static final String TEST_SERVER = "test_server";
    private static final String TEST_PROPERTY_PREFIX = "debezium.test.";
    private static final Logger log = LoggerFactory.getLogger(PGLog.class);


    @Override
    public void run(String... args) throws Exception {
        String offsetStorageFileFilenameConfig = "offset.storage.file.filename";
        Path offsetStorePath = Paths.get("./src/main/offset/", "connector-offsets.txt").toAbsolutePath();
        offsetStorePath.getParent().toFile().mkdirs();

        LoggingContext.forConnector(getClass().getSimpleName(), "debezium-engine", "engine");
        final Properties props = new Properties();
        props.putAll(getConfigMap());
        props.setProperty("name", "debezium-engine");
        props.setProperty("connector.class", "io.debezium.connector.postgresql.PostgresConnector");
        props.setProperty(offsetStorageFileFilenameConfig, offsetStorePath.toAbsolutePath().toString());
        props.setProperty("offset.flush.interval.ms", "0");
        props.setProperty("converter.schemas.enable", "false");

        DebeziumEngine<ChangeEvent<String, String>> engin = DebeziumEngine.create(Json.class).using(props)
                .notifying((records, committer) -> {
                    for (ChangeEvent<String, String> r : records) {
                        log.info("{}\n{}",r.key(), r.value());
//                        try {
//                            IdModel idModel = objectMapper.readValue(r.key(), IdModel.class);
//                            ChangeModel<User> userChangeModel = objectMapper.readValue(r.value(), new TypeReference<ChangeModel<User>>(){});
//                        } catch (JsonProcessingException e) {
//                            e.printStackTrace();
//                        }

                        committer.markProcessed(r);
                    }
                    committer.markBatchFinished();
                }).using(this.getClass().getClassLoader()).build();
        engin.run();
    }

    private Map<String,String> getConfigMap(){
        JdbcConfiguration jdbcConfiguration = JdbcConfiguration.copy(Configuration.fromSystemProperties("database."))
                .withDefault(JdbcConfiguration.DATABASE, StringUtils.defaultString(pgProperties.getDbName(),"postgres"))
                .withDefault(JdbcConfiguration.HOSTNAME, StringUtils.defaultString(pgProperties.getHost(),"192.168.56.10"))
                .withDefault(JdbcConfiguration.PORT, pgProperties.getPort() != null ? pgProperties.getPort() : 5432)
                .withDefault(JdbcConfiguration.USER, StringUtils.defaultString(pgProperties.getUsername(),"postgres"))
                .withDefault(JdbcConfiguration.PASSWORD, StringUtils.defaultString(pgProperties.getPassword(),"postgres"))
                .build();
        Configuration.Builder builder = Configuration.create();
        jdbcConfiguration.forEach((field, value) -> builder.with(DATABASE_CONFIG_PREFIX + field, value));
        builder.with(RelationalDatabaseConnectorConfig.SERVER_NAME, TEST_SERVER)
                .with(PostgresConnectorConfig.DROP_SLOT_ON_STOP, true)
                .with(PostgresConnectorConfig.STATUS_UPDATE_INTERVAL_MS, 100)
                .with(PostgresConnectorConfig.PLUGIN_NAME, decoderPlugin())
                .with(PostgresConnectorConfig.SSL_MODE, PostgresConnectorConfig.SecureConnectionMode.DISABLED);
        final String testNetworkTimeout = System.getProperty(TEST_PROPERTY_PREFIX + "network.timeout");
        if (testNetworkTimeout != null && testNetworkTimeout.length() != 0) {
            builder.with(PostgresConnectorConfig.STATUS_UPDATE_INTERVAL_MS, Integer.parseInt(testNetworkTimeout));
        }
        return builder.build().asMap();
    }

    private PostgresConnectorConfig.LogicalDecoder decoderPlugin() {
        final String s = System.getProperty(PostgresConnectorConfig.PLUGIN_NAME.name());
        return (s == null || s.length() == 0) ? PostgresConnectorConfig.LogicalDecoder.DECODERBUFS : PostgresConnectorConfig.LogicalDecoder.parse(s);
    }

}